package com.neo.sk.collector

import akka.actor.{Props, Actor}
import com.alibaba.fastjson.{TypeReference, JSON}
import com.neo.sk.common._
import com.neo.sk.domain.DoubanMovieDetail
import com.neo.sk.util.{ParseUtil, ParamsUtil}
import com.neo.sk.util.http.{InputStreamUtils, HttpClientUtil}
import models.Tables.MoSimpleDetailRow
import org.apache.http.impl.client.CloseableHttpClient
import org.apache.http.util.EntityUtils
import org.slf4j.LoggerFactory
import scala.concurrent.Future
import scala.concurrent.duration._
import scala.util.{Failure, Success}
import scala.collection.JavaConversions._
/**
 * User: Huangshanqi
 * Date: 2015/7/17
 * Time: 13:15
 */
class FetchMovieDetailActor(userClient: CloseableHttpClient)  extends Actor{
  private val logger = LoggerFactory.getLogger(this.getClass)
  import com.neo.sk.collector.Manager.myExecutionContext
  final val httpClientUtil = new HttpClientUtil()
  val saveActor = context.actorOf(Props[SaveActor](new SaveActor), "SaveActor-FetchMovieDetailActor-" + System.currentTimeMillis())
  private final val MaxRetryTime = 3

  override def receive: Receive = {

    case UrlTask(url) => {
//      logger.info("fetching url: " + url)
      val urlSegment = url.split("\\?")
      //http://api.douban.com/v2/movie/25723907
      val midPattern = "(.*)/movie/(\\d+)(.*)".r
      val midPattern(_,mid,_) = urlSegment(0)
      val retryTime = ParamsUtil.getRetryTimeFromUrl(urlSegment(1))

      if(retryTime < MaxRetryTime) {
        val f = Future {
          HttpClientUtil.executeHttpGet(url, userClient, null)
        }
        f.onComplete {
          case Success(response) => {
            try {
              val content = EntityUtils.toString(InputStreamUtils.getRealEntity(response.getEntity), "utf-8")
              val code = response.getStatusLine.getStatusCode
              logger.info("code============================= " + code)
              if (code == 200){
                if (content != null && content != "") {
//                  logger.info("result= " + content)
                  val movie = JSON.parseObject(content,Class.forName("com.neo.sk.domain.DoubanMovieDetail")).asInstanceOf[DoubanMovieDetail]
                  val movieDetailRow = MoSimpleDetailRow(
                  0,
                  mid.toLong,
                  if (movie.getRating!=null && movie.getRating.getAverage!=null) movie.getRating.getAverage else "",
                  if (movie.getRating!=null) movie.getRating.getNumRaters.toString else "",
                  if(movie.getTitle!=null) movie.getTitle else "",
                  if(movie.getSummary!=null)movie.getSummary else "",
                  if(movie.getAttrs.getLanguage!=null) movie.getAttrs.getLanguage.mkString(",") else "",
                  if(movie.getAttrs.getPubdate!=null) movie.getAttrs.getPubdate.mkString(",") else "",
                  if(movie.getAttrs.getCountry!=null) movie.getAttrs.getCountry.mkString(",") else "",
                  if(movie.getAttrs.getWriter!=null) movie.getAttrs.getWriter.mkString(",") else "",
                  if(movie.getAttrs.getDirector!=null) movie.getAttrs.getDirector.mkString(",") else "",
                  if(movie.getAttrs.getCast!=null) movie.getAttrs.getCast.mkString(",") else "",
                  if (movie.getAttrs.getMovie_duration!=null) movie.getAttrs.getMovie_duration.mkString(",") else "",
                  if(movie.getAttrs.getYear!=null) movie.getAttrs.getYear.mkString(",") else "",
                  if(movie.getAttrs.getMovie_type!=null) movie.getAttrs.getMovie_type.mkString(",") else "",
                  if(movie.getTags!=null) movie.getTags.map(tag => tag.getName + "#" +tag.getCount).mkString(",") else ""
                  )
                  saveActor ! SaveMovieDetail(movieDetailRow)

                  //finish
                  context.parent ! FinishFetchByTag(url + ",result=success")
                } else {
                  logger.info("code==== " + code + " with url " + url)
                  val retryUrl = urlSegment(0) + "?retry=" + (retryTime + 1)
                  self ! Retry(retryUrl)
                }
              }else{
                val retryUrl = urlSegment(0) + "?retry=" + (retryTime + 1)
                self ! Retry(retryUrl)
              }
              EntityUtils.consume(response.getEntity)
              response.close()
            }catch {
              case e:Exception =>
                logger.info(e.getMessage,e)
                val retryUrl = urlSegment(0) + "?retry=" + (retryTime + 1)
                self ! Retry(retryUrl)
            }
          }
          case Failure(e) =>{
            logger.info(e.getMessage,e)
            val retryUrl = urlSegment(0) + "?retry=" + (retryTime + 1)
            self ! Retry(retryUrl)
          }
        }
      }else{
        logger.info("get url:" + url + " more than " + MaxRetryTime +" times and fail")
        //finish
        context.parent ! FinishFetchByTag(url + ",result=failed")
      }
    }
    case Retry(url) =>{
      logger.info("retry =========== " + url)
      context.system.scheduler.scheduleOnce(10 seconds,self,UrlTask(url))
    }
    case unknow@_ => logger.info("unknow message: "+ unknow + "  in " + context.self.path.name + " from " + context.sender().path.name)
  }
}
